﻿using System;
using System.Collections.Generic;
using PhiStore.ObjectManaging;
using PhiStore.Packages;
using PhiStore.Packages.InternalPackages;
using PhiStore.Storage;
using PhiStore.Threading;

namespace PhiStore.Communication
{
    partial class Worker
    {
        private void Create(Create package)
        {
            if (PeerManager.Instance.PeerCount < (package.Policy.ReplicationNumber - 1))
            {
                SendReply(ReturnCode.UnableToSupportPolicy);
                SendReply("");
                return;
            }

            string key = package.Key;
            currentObjectName = key;
            ReturnCode code = ReturnCode.Failure;

            bool loop = true;
            while (loop) //generate key and negotiate for coordination role
            {
                loop = false;

                if (package.GenerateKey)
                    key = KeyHelper.GenerateObjectKey();

                PhiServer.Instance.ObjectManagerChannel.Send(new ObjectCreate(fromObjectManager, package, key));

                //receive reply from negotiator / object manager
            keepaliveloop:
                try
                {
                    code = (ReturnCode)fromObjectManager.Receive(1000);
                }
                catch (TimeoutException)
                {
                    SendReply(ReturnCode.WorkingOnObject);
                    goto keepaliveloop;
                }


                if (code == ReturnCode.ObjectAlreadyExist && package.GenerateKey)
                {
                    loop = true;
                }
                else if (code == ReturnCode.Success)
                { //perform negotiation for coordination role

                    SendReply(ReturnCode.WorkingOnObject); // keep client connection alive
                    ulong bullynumber;
                keepaliveloop2:
                    try
                    {
                        bullynumber = (ulong)fromObjectManager.Receive(1000);
                    }
                    catch (TimeoutException)
                    {
                        SendReply(ReturnCode.WorkingOnObject);
                        goto keepaliveloop2;
                    }


                    // a new key may be generated - that's OK
                    ReturnCode NegCode = Negotiate(key, bullynumber, package.Policy, package.Objects.Count, 0L);
                    if (NegCode.HasFlag(ReturnCode.NegotiationLost))
                    {
                        if (!package.GenerateKey)
                            code = ReturnCode.ObjectAlreadyExist;
                        loop = package.GenerateKey;
                    }
                    else if (NegCode.HasFlag(ReturnCode.NegotiationConflict))
                    {
                        loop = true;
                    }
                    else //success
                    {
                        loop = false;
                    }
                }
            } //generation and negotiation end

            if (code == ReturnCode.Success)
            {
                currentObjectName = key;
                ObjectLock oe = objectLock;

                //updating version
                oe.ObjectInformation = new ObjectInformation(oe.ObjectInformation.NextVersion, oe.ObjectInformation.Count, oe.ObjectInformation.FirstIndex);

                bool replicationSuccess = false;
                while (!replicationSuccess)
                {
                    //fetch a number of random peers
                    List<PeerContainer> peers = PeerManager.Instance.GetPeers(oe.Policy.ReplicationNumber - 1);
                    Channel<object> fromPeers = new Channel<object>(peers.Count);

                    if (peers.Count == 0 && oe.Policy.ReplicationNumber > 1) //not enough peers to support replication
                    {
                        code = ReturnCode.UnableToSupportPolicy;
                        replicationSuccess = true;
                    }
                    else
                    {
                        // create physical copies
                        code = oe.Device.CreateObject(key, oe.Policy, oe.ObjectInformation, oe.Policy.EnforceAtomicity);
                        for (long i = 0; i < oe.ObjectInformation.Count; i++)
                        {
                            code |= oe.Device.CreateObject(key, i, package.Objects[(int)i], oe.Policy.EnforceAtomicity);
                        }

                        oe.ReplicatingPeers = peers;
                        foreach (PeerContainer p in peers)
                        {
                            try
                            {
                                p.PeerWorkerChannel.Send
                                    (
                                    new ForwardedPackage(new ObjectCopy(key, package.Objects, oe.ObjectInformation), fromPeers)
                                    );

                                SendReply(ReturnCode.WorkingOnObject); // keep client connection alive
                                //receiving replies from peers
                                code |= (ReturnCode)fromPeers.Receive();
                            }
                            catch (PoisonException)
                            {
                                code |= ReturnCode.ConnectionLost;
                                PeerManager.Instance.RemovePeer(p);
                            }
                        }

                        if (oe.Policy.EnforceAtomicity)
                        {
                            PhiPackage abortCommit;
                            if (code == ReturnCode.Success)
                            {
                                abortCommit = new Commit(key);
                                oe.Device.Commit(key);
                                replicationSuccess = true;
                            }
                            else
                            {
                                abortCommit = new Abort(key);
                                oe.Device.Abort(key);
                                replicationSuccess = true;
                            }
                            ForwardedPackage fa = new ForwardedPackage(abortCommit, null);
                            foreach (PeerContainer p in peers)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                }
                                catch (PoisonException)
                                {
                                    PeerManager.Instance.RemovePeer(p);
                                }
                            }
                        }
                        else
                        {
                            replicationSuccess = true;
                        }
                    }
                }

                ReleaseLockEntry();

            }
            SendReply(code);
            SendReply(key);
        }

        private void ReadObject(Read package)
        {
            currentObjectName = package.Key;
            ReturnCode lockResult = LockEntry(); //obtaining lock

            if (lockResult == ReturnCode.Success)
            {
                ObjectLock oe = objectLock;

                long offset = package.Offset + oe.ObjectInformation.FirstIndex;

                if (oe.Coordinator && package.Offset + package.Count > oe.ObjectInformation.Count)
                {
                    ReleaseLockEntry();
                    SendReply(new ReadReply(ReturnCode.OutOfBounds, new object()));
                }
                else
                {
                    //used for fetching a copy
                    PeerContainer coordinator = oe.Policy.Prefetchable && !oe.Coordinator ? oe.ExternalCoordinator : null;
                    ObjectInformation oi = oe.ObjectInformation;


                    if (oe.Coordinator || //I'm coordinator
                        (oe.Policy.Prefetchable //object is prefetchable
                        && oi.Count >= package.Offset + package.Count) //I have the range
                    )
                    {
                        List<byte[]> objects = new List<byte[]>();

                        for (long i = 0L; i < package.Count; i++)
                        {
                            long index = i + offset;
                            byte[] objBytes = oe.Device.ReadObject(package.Key, index);
                            objects.Add(objBytes);
                        }
                        ReadReply reply = new ReadReply(lockResult, objects);

                        ReleaseLockEntry(); //releasing lock
                        SendReply(reply);
                    }
                    else
                    {
                        ReleaseLockEntry(); //releasing lock
                        SendReply(ReturnCode.WorkingOnObject); //keepalive
                        ForwardWait(package, oe.ExternalCoordinator);
                    }

                    if (coordinator != null && oi.Version == 0) //the object is prefetchable and I'm not the coordinator and I don't have a copy
                    {
                        try
                        {
                            oe.ExternalCoordinator.PeerWorkerChannel.Send(
                                new ForwardedPackage(new UpdateServerCopy(package.Key, oi), null)
                                );
                        }
                        catch (PoisonException)
                        {
                            //Coordinator is lost
                            PeerManager.Instance.RemovePeer(oe.ExternalCoordinator);
                        }
                    }
                }
            }
            else
            {
                SendReply(new ReadReply(lockResult, new object()));
            }
        }

        private void GetCount(ListCount package)
        {
            currentObjectName = package.Key;
            ReturnCode lockResult = LockEntry(); //obtaining lock
            if (lockResult == ReturnCode.Success)
            {
                if (objectLock.Coordinator)
                {
                    ReadReply reply = new ReadReply(lockResult, objectLock.ObjectInformation.Count);
                    ReleaseLockEntry();
                    SendReply(reply);
                }
                else
                {//FORWARD TO COORDINATOR
                    ObjectLock ol = objectLock;
                    ReleaseLockEntry();
                    ForwardWait(package, ol.ExternalCoordinator);
                }
            }
            else
            {
                SendReply(new ReadReply(lockResult, new object()));
            }

        }

        private void UpdateObject(Update package)
        {
            currentObjectName = package.Key;
            ReturnCode lockResult = LockEntry(); //obtaining lock

            if (lockResult == ReturnCode.Success)
            {
                ObjectLock oe = objectLock;

                if (!oe.Policy.MayModify)
                {
                    ReleaseLockEntry();
                    SendReply(ReturnCode.PolicyViolated);
                }
                else if (oe.Coordinator && package.Offset + package.Objects.Count > oe.ObjectInformation.Count)
                {
                    ReleaseLockEntry();
                    SendReply(ReturnCode.OutOfBounds);
                }
                else
                {
                    if (oe.Coordinator)
                    {
                        ReturnCode result = lockResult;
                        string key = package.Key;

                        ObjectInformation newInfo = new ObjectInformation(oe.ObjectInformation.NextVersion,
                                                    oe.ObjectInformation.Count,
                                                    oe.ObjectInformation.FirstIndex);

                        SendReply(ReturnCode.WorkingOnObject); //keepalive
                        #region Local update
                        for (long i = 0; i < package.Objects.Count; i++)
                        {
                            byte[] data = package.Objects[(int)i];
                            long index = package.Offset + oe.ObjectInformation.FirstIndex + i;
                            result |= oe.Device.UpdateObject(package.Key, index, data,
                                oe.Policy.EnforceAtomicity
                                );
                        }


                        result |= oe.Device.UpdateObjectInfo(
                            package.Key,
                            newInfo,
                            oe.Policy.EnforceAtomicity
                            );
                        ObjectInformation oldInfo = oe.ObjectInformation;
                        oe.ObjectInformation = newInfo;
                        #endregion

                        Channel<object> fromPeers = new Channel<object>(1);

                        UpdateCopy uc = new UpdateCopy(key, newInfo, package.Offset, package.Objects);
                        ForwardedPackage fa = new ForwardedPackage(uc, fromPeers);
                        List<PeerContainer> workingPeers = new List<PeerContainer>();

                        List<PeerContainer> peers = oe.ReplicatingPeers;
                        int stop = Math.Max(peers.Count,(int) oe.Policy.ReplicationNumber - 1);
                        for (int i = 0; i < stop; i++)
                        {
                            PeerContainer p = i >= peers.Count ? null : peers[i];
                            object reply = new object();
                            if (p != null)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                    SendReply(ReturnCode.WorkingOnObject);
                                    reply = fromPeers.Receive();
                                }
                                catch (PoisonException)
                                {
                                    reply = ReturnCode.ConnectionLost;
                                }
                            }
                            if (isConnectionLost(reply) || p == null)
                            {
                                //this replication failed
                                if (p == null)
                                    PeerManager.Instance.RemovePeer(p);

                                PeerContainer newpeer = PeerManager.Instance.GetNewPeer(oe.ReplicatingPeers);

                                if (newpeer == null) //unable to support replication
                                {
                                    oe.Device.Abort(key); //abort local modifications
                                    oe.ObjectInformation = oldInfo;

                                    foreach (PeerContainer peer in workingPeers) //abort modifications done so far on peers
                                    {
                                        try
                                        {
                                            peer.PeerWorkerChannel.Send(new ForwardedPackage(new Abort(key), null));
                                        }
                                        catch (PoisonException) { }
                                    }
                                    ReleaseLockEntry();
                                    SendReply(ReturnCode.UnableToSupportPolicy);
                                    return;
                                }

                                //add a new one and create a copy on that peer
                                result |= CreateNewCopy(newpeer);
                                workingPeers.Add(newpeer);
                            }
                            else
                            {
                                result |= (ReturnCode)reply;
                                workingPeers.Add(p);
                            }
                        }
                        oe.ReplicatingPeers = workingPeers;

                        if (oe.Policy.EnforceAtomicity)
                        {
                            PhiPackage abortCommit;
                            if (result == ReturnCode.Success)
                            {
                                abortCommit = new Commit(key);
                                oe.Device.Commit(key);
                            }
                            else
                            {
                                abortCommit = new Abort(key);
                                oe.Device.Abort(key);
                            }
                            fa = new ForwardedPackage(abortCommit, null);
                            foreach (PeerContainer p in oe.ReplicatingPeers)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                }
                                catch (PoisonException)
                                {
                                    PeerManager.Instance.RemovePeer(p);
                                }
                            }
                        }
                        ReleaseLockEntry();

                        SendReply(result);
                    }
                    else
                    {
                        //forward request
                        SendReply(ReturnCode.WorkingOnObject);
                        ReleaseLockEntry();

                        ForwardWait(package, oe.ExternalCoordinator);
                    }
                }
            }
            else
            {
                SendReply(lockResult);
            }
        }

        private void Append(ListAppend package)
        {
            currentObjectName = package.Key;
            string key = package.Key;
            ReturnCode lockResult = LockEntry();

            if (lockResult == ReturnCode.Success)
            {
                ObjectLock oe = objectLock;

                if (!oe.Policy.MayAppend)
                {
                    ReleaseLockEntry();
                    SendReply(ReturnCode.PolicyViolated);
                }
                else
                {
                    if (oe.Coordinator)
                    {
                        ReturnCode result = lockResult;

                        ObjectInformation newInfo = new ObjectInformation(oe.ObjectInformation.NextVersion,
                                                    oe.ObjectInformation.Count + package.Objects.Count,
                                                    oe.ObjectInformation.FirstIndex);

                        SendReply(ReturnCode.WorkingOnObject); //keepalive
                        #region Local update
                        for (long i = 0; i < package.Objects.Count; i++)
                        {
                            byte[] data = package.Objects[(int)i];
                            long index = oe.ObjectInformation.Count + oe.ObjectInformation.FirstIndex + i;
                            result |= oe.Device.CreateObject(key, index, data,
                                oe.Policy.EnforceAtomicity
                                );
                        }


                        result |= oe.Device.UpdateObjectInfo(
                            key,
                            newInfo,
                            oe.Policy.EnforceAtomicity
                            );
                        ObjectInformation oldInfo = oe.ObjectInformation;
                        oe.ObjectInformation = newInfo;
                        #endregion

                        Channel<object> fromPeers = new Channel<object>(1);

                        UpdateCopy uc = new UpdateCopy(key, newInfo, oe.ObjectInformation.Count, package.Objects);
                        ForwardedPackage fa = new ForwardedPackage(uc, fromPeers);
                        List<PeerContainer> workingPeers = new List<PeerContainer>();

                        List<PeerContainer> peers = oe.ReplicatingPeers;
                        int stop = Math.Max(peers.Count, (int)oe.Policy.ReplicationNumber - 1);
                        for (int i = 0; i < stop; i++)
                        {
                            PeerContainer p = i >= peers.Count ? null : peers[i];
                            object reply = new object();
                            if (p != null)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                    SendReply(ReturnCode.WorkingOnObject);
                                    reply = fromPeers.Receive();
                                }
                                catch (PoisonException)
                                {
                                    reply = ReturnCode.ConnectionLost;
                                }
                            }
                            if (isConnectionLost(reply) || p == null)
                            {
                                //this replication failed
                                if (p == null)
                                    PeerManager.Instance.RemovePeer(p);

                                PeerContainer newpeer = PeerManager.Instance.GetNewPeer(oe.ReplicatingPeers);

                                if (newpeer == null) //unable to support replication
                                {
                                    oe.Device.Abort(key); //abort local modifications
                                    oe.ObjectInformation = oldInfo;

                                    foreach (PeerContainer peer in workingPeers) //abort modifications done so far on peers
                                    {
                                        try
                                        {
                                            peer.PeerWorkerChannel.Send(new ForwardedPackage(new Abort(key), null));
                                        }
                                        catch (PoisonException) { }
                                    }
                                    ReleaseLockEntry();
                                    SendReply(ReturnCode.UnableToSupportPolicy);
                                    return;
                                }

                                //add a new one and create a copy on that peer
                                result |= CreateNewCopy(newpeer);
                                workingPeers.Add(newpeer);
                            }
                            else
                            {
                                result |= (ReturnCode)reply;
                                workingPeers.Add(p);
                            }
                        }
                        oe.ReplicatingPeers = workingPeers;

                        if (oe.Policy.EnforceAtomicity)
                        {
                            PhiPackage abortCommit;
                            if (result == ReturnCode.Success)
                            {
                                abortCommit = new Commit(key);
                                oe.Device.Commit(key);
                            }
                            else
                            {
                                abortCommit = new Abort(key);
                                oe.Device.Abort(key);
                            }
                            fa = new ForwardedPackage(abortCommit, null);
                            foreach (PeerContainer p in oe.ReplicatingPeers)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                }
                                catch (PoisonException)
                                {
                                    PeerManager.Instance.RemovePeer(p);
                                }
                            }
                        }
                        ReleaseLockEntry();

                        SendReply(result);
                    }
                    else
                    {
                        //forward request
                        SendReply(ReturnCode.WorkingOnObject);
                        ReleaseLockEntry();

                        ForwardWait(package, oe.ExternalCoordinator);
                    }
                }
            }
            else
            {
                SendReply(lockResult);
            }

        }

        private void DeleteListEntries(ListDelete package)
        {
            currentObjectName = package.Key;

            ReturnCode lockResult = LockEntry();

            if (lockResult == ReturnCode.Success)
            {
                ObjectLock oe = objectLock;

                if (!oe.Policy.MayDelete)
                {
                    ReleaseLockEntry();
                    SendReply(ReturnCode.PolicyViolated);
                }
                else if (package.Offset + package.Count > oe.ObjectInformation.Count)
                {
                    ReleaseLockEntry();
                    SendReply(ReturnCode.OutOfBounds);
                }
                else
                {
                    ReturnCode result = lockResult;

                    if (oe.Coordinator)
                    {
                        long startDeletion = oe.ObjectInformation.FirstIndex + package.Offset;

                        for (long i = 0; i < package.Count; i++)
                        {
                            long index = i + startDeletion;
                            oe.Device.DeleteObject(package.Key, index, oe.Policy.EnforceAtomicity);
                        }

                        ObjectInformation newInfo;

                        long leftPortionSize = package.Offset;
                        long rightPortionSize = oe.ObjectInformation.Count - package.Count - package.Offset;

                        if (leftPortionSize < rightPortionSize)
                        {   //first half is smallest
                            for (long index = startDeletion - 1; index >= oe.ObjectInformation.FirstIndex; index--)
                            {
                                long newIndex = index + package.Count;
                                oe.Device.Rename(package.Key, index, newIndex, oe.Policy.EnforceAtomicity);
                            }

                            newInfo = new ObjectInformation(oe.ObjectInformation.NextVersion,
                                oe.ObjectInformation.Count - package.Count,
                               oe.ObjectInformation.FirstIndex + package.Count);

                        }
                        else
                        { //second half is smallest
                            for (long i = 0; i < oe.ObjectInformation.Count - package.Offset - package.Count; i++)
                            {
                                long oldIndex = startDeletion + i + package.Count;
                                long newIndex = startDeletion + i;
                                oe.Device.Rename(package.Key, oldIndex, newIndex, oe.Policy.EnforceAtomicity);

                            }

                            newInfo = new ObjectInformation(oe.ObjectInformation.NextVersion,
                                oe.ObjectInformation.Count - package.Count,
                                oe.ObjectInformation.FirstIndex);
                        }

                        result |= oe.Device.UpdateObjectInfo(package.Key, newInfo, oe.Policy.EnforceAtomicity);

                        ObjectInformation oldinfo = oe.ObjectInformation;
                        oe.ObjectInformation = newInfo;

                        Channel<object> fromPeers = new Channel<object>(1);

                        ListDeleteCopy ldc = new ListDeleteCopy(package.Key, package.Offset, package.Count, newInfo);
                        ForwardedPackage fa = new ForwardedPackage(ldc, fromPeers);
                        List<PeerContainer> workingPeers = new List<PeerContainer>();

                        List<PeerContainer> peers = oe.ReplicatingPeers;
                        int stop = Math.Max(peers.Count, (int)oe.Policy.ReplicationNumber - 1);
                        for (int i = 0; i < stop; i++)
                        {
                            PeerContainer p = i >= peers.Count ? null : peers[i];
                            object reply = new object();
                            if (p != null)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                    SendReply(ReturnCode.WorkingOnObject);
                                    reply = fromPeers.Receive();
                                }
                                catch (PoisonException)
                                {
                                    reply = ReturnCode.ConnectionLost;
                                    PeerManager.Instance.RemovePeer(p);
                                }
                            }
                            if (isConnectionLost(reply) || p == null)
                            {
                                //this replication failed
                                if (p == null)
                                    PeerManager.Instance.RemovePeer(p);

                                PeerContainer newpeer = PeerManager.Instance.GetNewPeer(oe.ReplicatingPeers);

                                if (newpeer == null) //unable to support replication
                                {
                                    oe.Device.Abort(package.Key); //abort local modifications
                                    oe.ObjectInformation = oldinfo;

                                    foreach (PeerContainer peer in workingPeers) //abort modifications done so far on peers
                                    {
                                        try
                                        {
                                            peer.PeerWorkerChannel.Send(new ForwardedPackage(new Abort(package.Key), null));
                                        }
                                        catch (PoisonException) { }
                                    }
                                    ReleaseLockEntry();
                                    SendReply(ReturnCode.UnableToSupportPolicy);
                                    return;
                                }

                                //add a new one and create a copy on that peer
                                result |= CreateNewCopy(newpeer);
                                workingPeers.Add(newpeer);
                            }
                            else
                            {
                                result |= (ReturnCode)reply;
                                workingPeers.Add(p);
                            }
                        }
                        oe.ReplicatingPeers = workingPeers;

                        if (oe.Policy.EnforceAtomicity)
                        {
                            PhiPackage abortCommit;
                            if (result == ReturnCode.Success)
                            {
                                abortCommit = new Commit(package.Key);
                                oe.Device.Commit(package.Key);
                            }
                            else
                            {
                                abortCommit = new Abort(package.Key);
                                oe.Device.Abort(package.Key);
                            }
                            fa = new ForwardedPackage(abortCommit, null);
                            foreach (PeerContainer p in oe.ReplicatingPeers)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                }
                                catch (PoisonException)
                                {
                                    PeerManager.Instance.RemovePeer(p);
                                }
                            }
                        }
                        ReleaseLockEntry();

                        SendReply(result);
                    }
                    else
                    {
                        ReleaseLockEntry();

                        SendReply(ReturnCode.WorkingOnObject);
                        ForwardWait(package, oe.ExternalCoordinator);
                    }
                }


            }
            else
            {//no lock aqcuired
                SendReply(lockResult);
            }
        }

        private void Pop(ListPop package)
        {
            currentObjectName = package.Key;
            ReturnCode lockResult = LockEntry(); //obtaining lock
            if (lockResult == ReturnCode.Success)
            {
                ObjectLock oe = objectLock;
                if (!oe.Policy.MayPop)
                {
                    ReleaseLockEntry();
                    SendReply(new ReadReply(ReturnCode.PolicyViolated, new object()));
                }
                else
                {
                    ReturnCode result = lockResult;
                    if (oe.Coordinator)
                    {
                        long index = oe.ObjectInformation.FirstIndex;

                        byte[] data = oe.Device.ReadObject(package.Key, index);

                        ObjectInformation newInfo = new ObjectInformation(
                            oe.ObjectInformation.NextVersion,
                            oe.ObjectInformation.Count - 1,
                            oe.ObjectInformation.FirstIndex + 1);

                        result |= oe.Device.DeleteObject(package.Key, index, oe.Policy.EnforceAtomicity);
                        result |= oe.Device.UpdateObjectInfo(package.Key, newInfo, oe.Policy.EnforceAtomicity);

                        ObjectInformation oldInfo = oe.ObjectInformation;
                        oe.ObjectInformation = newInfo;

                        Channel<object> fromPeers = new Channel<object>(1);

                        ListDeleteCopy ldc = new ListDeleteCopy(package.Key, 0, 1, newInfo);
                        ForwardedPackage fa = new ForwardedPackage(ldc, fromPeers);
                        List<PeerContainer> workingPeers = new List<PeerContainer>();

                        List<PeerContainer> peers = oe.ReplicatingPeers;
                        int stop = Math.Max(peers.Count, (int)oe.Policy.ReplicationNumber - 1);
                        for (int i = 0; i < stop; i++)
                        {
                            PeerContainer p = i >= peers.Count ? null : peers[i];
                            object reply = new object();
                            if (p != null)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                    SendReply(ReturnCode.WorkingOnObject);
                                    reply = fromPeers.Receive();
                                }
                                catch (PoisonException)
                                {
                                    PeerManager.Instance.RemovePeer(p);
                                    reply = ReturnCode.ConnectionLost;
                                }
                            }
                            if (isConnectionLost(reply) || p == null)
                            {
                                //this replication failed
                                if (p == null)
                                    PeerManager.Instance.RemovePeer(p);

                                PeerContainer newpeer = PeerManager.Instance.GetNewPeer(oe.ReplicatingPeers);

                                if (newpeer == null) //unable to support replication
                                {
                                    oe.Device.Abort(package.Key); //abort local modifications
                                    oe.ObjectInformation = oldInfo;

                                    foreach (PeerContainer peer in workingPeers) //abort modifications done so far on peers
                                    {
                                        try
                                        {
                                            peer.PeerWorkerChannel.Send(new ForwardedPackage(new Abort(package.Key), null));
                                        }
                                        catch (PoisonException) { }
                                    }
                                    ReleaseLockEntry();
                                    SendReply(new ReadReply(ReturnCode.UnableToSupportPolicy, new object()));
                                    return;
                                }

                                //add a new one and create a copy on that peer
                                result |= CreateNewCopy(newpeer);
                                workingPeers.Add(newpeer);
                            }
                            else
                            {
                                result |= (ReturnCode)reply;
                                workingPeers.Add(p);
                            }
                        }
                        oe.ReplicatingPeers = workingPeers;

                        if (oe.Policy.EnforceAtomicity)
                        {
                            PhiPackage abortCommit;
                            if (result == ReturnCode.Success)
                            {
                                abortCommit = new Commit(package.Key);
                                oe.Device.Commit(package.Key);
                            }
                            else
                            {
                                abortCommit = new Abort(package.Key);
                                oe.Device.Abort(package.Key);
                            }
                            fa = new ForwardedPackage(abortCommit, null);
                            foreach (PeerContainer p in oe.ReplicatingPeers)
                            {
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                }
                                catch (PoisonException)
                                {
                                    PeerManager.Instance.RemovePeer(p);
                                }
                            }
                        }
                        ReleaseLockEntry();

                        SendReply(new ReadReply(result, data));
                    }
                    else
                    { //forward request
                        ReleaseLockEntry();
                        SendReply(ReturnCode.WorkingOnObject);
                        ForwardWait(package, oe.ExternalCoordinator);
                    }
                }

            }
            else
            { //lock wasn't obtained
                SendReply(new ReadReply(lockResult, new object()));
            }
        }

        private void DeleteObject(Delete deleteAction)
        {
            currentObjectName = deleteAction.Key;
            ReturnCode result = LockEntry();

            if (result == ReturnCode.Success)
            {
                ObjectLock oe = (ObjectLock)objectLock;
                if (!oe.Policy.MayDelete) //Deletion is not permitted
                {
                    ReleaseLockEntry();
                    SendReply(ReturnCode.PolicyViolated);
                }
                else
                {
                    if (oe.Coordinator)
                    {
                        result |= oe.Device.DeleteObject(deleteAction.Key, oe.Policy.EnforceAtomicity);
                        if (result == ReturnCode.Success)
                        {
                            Channel<object> fromPeers = new Channel<object>(1);

                            DeleteCopy dc = new DeleteCopy(deleteAction.Key);
                            ForwardedPackage fa = new ForwardedPackage(dc, fromPeers);

                            foreach (PeerContainer p in PeerManager.Instance.GetPeers())
                            {
                                object reply;
                                try
                                {
                                    p.PeerWorkerChannel.Send(fa);
                                    SendReply(ReturnCode.WorkingOnObject);
                                    reply = fromPeers.Receive();
                                }
                                catch (PoisonException)
                                {
                                    reply = ReturnCode.ConnectionLost;
                                }

                                if (!isConnectionLost(reply)) //if connection is lost, it's ok
                                {
                                    result |= (ReturnCode)reply;
                                }
                            }

                            if (oe.Policy.EnforceAtomicity)
                            {
                                SendReply(ReturnCode.WorkingOnObject);
                                PhiPackage abortCommit;
                                if (result == ReturnCode.Success)
                                {
                                    abortCommit = new Commit(deleteAction.Key);
                                    oe.Device.Commit(deleteAction.Key);
                                }
                                else
                                {
                                    abortCommit = new Abort(deleteAction.Key);
                                    oe.Device.Abort(deleteAction.Key);
                                }
                                fa = new ForwardedPackage(abortCommit, null);
                                foreach (PeerContainer p in PeerManager.Instance.GetPeers())
                                {
                                    try
                                    {
                                        p.PeerWorkerChannel.Send(fa);
                                    }
                                    catch (PoisonException)
                                    {
                                        PeerManager.Instance.RemovePeer(p);
                                        //connection is marked as lost
                                    }
                                }
                            }
                            PhiServer.Instance.ObjectManagerChannel.Send(new RemoveObject(deleteAction.Key, fromObjectManager));

                        keepaliveloop:
                            try
                            {
                                result |= (ReturnCode)fromObjectManager.Receive(1000);
                            }
                            catch (TimeoutException)
                            {
                                SendReply(ReturnCode.WorkingOnObject);
                                goto keepaliveloop;
                            }
                        }
                        SendReply(result);
                    }
                    else
                    {
                        //forward request
                        ReleaseLockEntry();
                        SendReply(ReturnCode.WorkingOnObject);

                        ForwardWait(deleteAction, oe.ExternalCoordinator);
                    }
                }

            }
            else
            { // no lock obtained
                SendReply(result);
            }
        }

        private void ForwardWait(PhiPackage action, PeerContainer Coordinator)
        {
            Channel<object> fromCoordinator = new Channel<object>(1);

            bool loop = true;
            object o = ReturnCode.Failure;
            if (Coordinator == null)
            {
                Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "Coordinator is null for " + action.ToString());
                o = ReturnCode.ConnectionLost;
            }
            else
            {
                try
                {
                    Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "Forwarding " + action.ToString() + " to " + Coordinator.BC.ToString());
                    Coordinator.PeerWorkerChannel.Send(new ForwardedPackage(action, fromCoordinator));
                }
                catch (PoisonException)
                {
                    loop = false;
                    o = ReturnCode.ConnectionLost;
                }

                while (loop)
                {
                    o = fromCoordinator.Receive();
                    
                    if (!ReturnCode.WorkingOnObject.Equals(o))
                        loop = false;

                    SendReply(ReturnCode.WorkingOnObject);
                }
            }

            if (isConnectionLost(o)) //connection to coordinator lost
            {
                SendReply(ReturnCode.WorkingOnObject);
                bool loopNegotiation = PeerManager.Instance.RemovePeer(Coordinator);


                if (!loopNegotiation)
                {
                    SendReply(ReturnCode.Failure);
                    return;
                }

                ReturnCode NegCode = ReturnCode.Success;
                while (loopNegotiation)
                {
                    //we'll negotiate for coordination role
                    PhiServer.Instance.ObjectManagerChannel.Send(
                        new NegotiationInitialization(fromObjectManager, currentObjectName)
                        );

                    ReturnCode negInit;
                keepaliveloop:
                    try
                    {
                        negInit = (ReturnCode)fromObjectManager.Receive(1000);
                    }
                    catch (TimeoutException)
                    {
                        SendReply(ReturnCode.WorkingOnObject);
                        goto keepaliveloop;
                    }


                    if (negInit == ReturnCode.Success)
                    {
                        ulong bullynumber;
                    keepaliveloop2:
                        try
                        {
                            bullynumber = (ulong)fromObjectManager.Receive(1000);
                        }
                        catch (TimeoutException)
                        {
                            SendReply(ReturnCode.WorkingOnObject);
                            goto keepaliveloop2;
                        }


                        ObjectLock oe;
                    keepaliveloop3:
                        try
                        {
                            oe = (ObjectLock)fromObjectManager.Receive(1000);

                        }
                        catch (TimeoutException)
                        {
                            SendReply(ReturnCode.WorkingOnObject);
                            goto keepaliveloop3;
                        }

                        objectLock = oe;

                        NegCode = Negotiate(currentObjectName, bullynumber, oe.Policy, oe.ObjectInformation.Count, oe.ObjectInformation.Version);

                        if (NegCode.HasFlag(ReturnCode.NegotiationConflict))
                            loopNegotiation = true;
                        else
                            loopNegotiation = false; //I either lost or won
                    }
                }


                if (NegCode == ReturnCode.Success &&  //I won 
                    objectLock.ObjectInformation.Version > 0) //and a copy was fetched / or I had it already
                {
                    //maintain replication
                    if (PeerManager.Instance.PeerCount - 1 < objectLock.Policy.ReplicationNumber)
                    {
                        //unable to support policy - failing here!
                    }
                    else
                    {
                        List<PeerContainer> peers = PeerManager.Instance.GetPeers(objectLock.Policy.ReplicationNumber - 1);
                        Channel<object> fromPeers = new Channel<object>(peers.Count);

                        List<byte[]> objects = new List<byte[]>();

                        long offset = objectLock.ObjectInformation.FirstIndex;
                        for (long i = 0; i < objectLock.ObjectInformation.Count; i++)
                        {
                            long index = i + offset;
                            objects.Add(objectLock.Device.ReadObject(currentObjectName, index));
                        }

                        foreach (PeerContainer peer in peers)
                        {
                            ReturnCode code = ReturnCode.Failure;
                            try
                            {
                                peer.PeerWorkerChannel.Send(
                                    new ForwardedPackage(new ObjectCopy(currentObjectName, objects, objectLock.ObjectInformation), fromPeers)
                                    );
                                SendReply(ReturnCode.WorkingOnObject); // keep client connection alive
                                code = (ReturnCode)fromPeers.Receive();
                            }
                            catch (PoisonException)
                            {
                                code |= ReturnCode.ConnectionLost;
                            }
                            if (code == ReturnCode.Success)
                                objectLock.ReplicatingPeers.Add(peer); //add to replicating peers

                            if (objectLock.Policy.EnforceAtomicity)
                            {
                                PhiPackage abortCommit;
                                if (code == ReturnCode.Success)
                                {
                                    abortCommit = new Commit(currentObjectName);
                                }
                                else
                                {
                                    abortCommit = new Abort(currentObjectName);
                                }
                                ForwardedPackage fa = new ForwardedPackage(abortCommit, null);
                                foreach (PeerContainer p in peers)
                                {
                                    try
                                    {
                                        p.PeerWorkerChannel.Send(fa);
                                    }
                                    catch (PoisonException)
                                    {
                                        PeerManager.Instance.RemovePeer(p);
                                    }
                                }
                            }
                        }
                    }
                }


                if (NegCode == ReturnCode.Success &&  //I won negotiation
                    objectLock.ObjectInformation.Version == 0L) // but no peer had a copy
                {
                    //OBJECT IS LOST
                    Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), currentObjectName + " is lost... SORRY!!!!!");
                    Channel<object> fromPeers = new Channel<object>(1);

                    DeleteCopy dc = new DeleteCopy(currentObjectName);
                    ForwardedPackage fa = new ForwardedPackage(dc, fromPeers);

                    foreach (PeerContainer p in PeerManager.Instance.GetPeers())
                    {
                        object reply;
                        try
                        {
                            p.PeerWorkerChannel.Send(fa);
                            SendReply(ReturnCode.WorkingOnObject);
                            reply = fromPeers.Receive();
                        }
                        catch (PoisonException)
                        {
                            reply = ReturnCode.ConnectionLost;
                        }

                        if (!isConnectionLost(reply)) //if connection is lost, it's ok
                        {
                            NegCode |= (ReturnCode)reply;
                        }
                    }

                    objectLock.Device.DeleteObject(currentObjectName, false);

                    if (objectLock.Policy.EnforceAtomicity)
                    {
                        PhiPackage abortCommit;
                        if (NegCode == ReturnCode.Success)
                        {
                            abortCommit = new Commit(currentObjectName);
                        }
                        else
                        {
                            abortCommit = new Abort(currentObjectName);
                        }
                        fa = new ForwardedPackage(abortCommit, null);
                        foreach (PeerContainer p in PeerManager.Instance.GetPeers())
                        {
                            try
                            {
                                p.PeerWorkerChannel.Send(fa);
                            }
                            catch (PoisonException)
                            {
                                //connection is lost
                                PeerManager.Instance.RemovePeer(p);
                            }
                        }
                    }
                    PhiServer.Instance.ObjectManagerChannel.Send(new RemoveObject(currentObjectName, fromObjectManager));
                keepaliveloop:
                    try
                    {
                        NegCode |= (ReturnCode)fromObjectManager.Receive(1000);
                    }
                    catch (TimeoutException)
                    {
                        SendReply(ReturnCode.WorkingOnObject);
                        goto keepaliveloop;
                    }
                }


                ReleaseLockEntry();
                Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "Negotiation completed. Redoing: " + action.ToString());
                PerformAction(action);//redo package
                //SendReply(ReturnCode.ObjectIsBusy); //let client retry package
            }
            else
            {
                SendReply(o);
            }
        }

        private ReturnCode CreateNewCopy(PeerContainer newpeer)
        {
            ObjectLock oe = objectLock;

            Channel<object> frompeer = new Channel<object>(1);

            List<byte[]> objs = new List<byte[]>();

            long offset = oe.ObjectInformation.FirstIndex;
            for (long i = 0L; i < oe.ObjectInformation.Count; i++)
            {
                long index = i + offset;
                byte[] objBytes = oe.Device.ReadObject(currentObjectName, index);
                objs.Add(objBytes);
            }

            ObjectInformation oi = oe.ObjectInformation;

            ForwardedPackage action = new ForwardedPackage(
                new ObjectCopy(currentObjectName, objs, oi),
                frompeer);
            try
            {
                newpeer.PeerWorkerChannel.Send(action);

                return (ReturnCode)frompeer.Receive();
            }
            catch (PoisonException)
            {
                return ReturnCode.ConnectionLost;
            }
        }

        private void TryFetchCopyFromPeer(PeerContainer peer)
        {
            Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "Fetching a copy of " + currentObjectName + " from " + peer.BC.ToString());
            if (peer == null)
                Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "peer is null");
            if (peer.PeerWorkerChannel == null)
                Log.Instance.AddLogEntry(connection.Client.RemoteEndPoint.ToString(), "peer Action channel is null");

            ObjectLock oe = objectLock;

            Channel<object> fromPeer = new Channel<object>(1);

            GetObjectInformation g = new GetObjectInformation(currentObjectName);
            object reply;
            try
            {
                peer.PeerWorkerChannel.Send(new ForwardedPackage(g, fromPeer));
                reply = fromPeer.Receive();
            }
            catch (PoisonException)
            {
                reply = ReturnCode.ConnectionLost;
            }

            if (!isConnectionLost(reply)) //oh snap
            {
                ObjectInformation oi = (ObjectInformation)reply;

                if (oi.Version > oe.ObjectInformation.Version) //I don't have the latest copy
                {
                    ReturnCode gotCopy = ReturnCode.Success;

                    for (long i = 0; i < oi.Count; i++)
                    {
                        long index = oi.FirstIndex + i;
                        ForwardedPackage getIndex = new ForwardedPackage(
                            new GetData(currentObjectName, index),
                            fromPeer);
                        object read;
                        try
                        {
                            peer.PeerWorkerChannel.Send(getIndex);
                            read = fromPeer.Receive();
                        }
                        catch (PoisonException)
                        {
                            read = ReturnCode.ConnectionLost;
                        }

                        if (!isConnectionLost(read) && read is ReadReply)
                        {
                            ReadReply readreply = (ReadReply)read;
                            gotCopy |= readreply.Code;

                            if (readreply.Code == ReturnCode.Success)
                            {
                                if (oe.Device.Exist(currentObjectName, index))
                                {
                                    gotCopy |= oe.Device.UpdateObject(currentObjectName, index, readreply.Data<byte[]>(), false);
                                }
                                else
                                {
                                    gotCopy |= oe.Device.CreateObject(currentObjectName, index, readreply.Data<byte[]>(), false);
                                }
                            }
                        }
                        else
                        {
                            gotCopy |= ReturnCode.ConnectionLost;
                        }
                    }

                    if (gotCopy == ReturnCode.Success)
                    {
                        oe.ObjectInformation = oi;
                        oe.Device.UpdateObjectInfo(currentObjectName, oi, false);
                    }
                }

            }
        }


    }
}
